/*
 *  Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  See the NOTICE file distributed with
 *  this work for additional information regarding copyright ownership.
 *  The ASF licenses this file to You under the Apache License, Version 2.0
 *  (the "License"); you may not use this file except in compliance with
 *  the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package org.apache.tomcat.util.threads;

import java.util.Collection;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;

import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;

/**
 * Shared latch that allows the latch to be acquired a limited number of times
 * after which all subsequent requests to acquire the latch will be placed in a
 * FIFO queue until one of the shares is returned.
 */
public class LimitLatch {

    private static final Log log = LogFactory.getLog(LimitLatch.class);
    //AQS（抽象队列式同步器）其实就是一个框架，实现它的类可以自定义控制线程什么时候挂起什么时候释放（参考：https://blog.csdn.net/weixin_40391011/article/details/104701988）
    private class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1L;

        public Sync() {
        }
        //定义了一旦当前连接数大于了设置的最大连接数，那么就会返回 -1表示将此线程放入AQS队列中等待。
        @Override
        protected int tryAcquireShared(int ignored) {
            //定义了AtomicLong类型的count数量，每次获取锁之后会加1
            long newCount = count.incrementAndGet();
            //如果count大于最大的限制并且released标识为false(默认为false,在调用relaseAll的时候会置为true)
            if (!released && newCount > limit) {
                // Limit exceeded
                //原子性的减1，还原
                count.decrementAndGet();
                //这里返回-1 返回-1的时候就会将当前线程放入CLH队列中等待被唤醒
                return -1;
            } else {
                //返回1 则当前线程不需要被挂起
                return 1;
            }
        }
        // 释放资源
        @Override
        protected boolean tryReleaseShared(int arg) {
            //释放锁的时候count数量减1
            count.decrementAndGet();
            return true;
        }
    }

    private final Sync sync;
    private final AtomicLong count;
    private volatile long limit;
    private volatile boolean released = false;

    /**
     * Instantiates a LimitLatch object with an initial limit.
     * @param limit - maximum number of concurrent acquisitions of this latch
     *1：LimitLatch内部定义了内部类Sync，而Sync扩展了AQS（在 Java API 文档中标明 AQS 的子类应定义为非公共内部帮助类，用于实现其封闭类的同步属性）。
     *AQS 是分为独占和共享两种模式，当以独占模式获取时，其他线程尝试的获取无法成功。多个线程获取共享模式可能（但不一定）成功，LimitLatch 的内部类 Sync 实现的抽象方法则属于共享模式。
     *在独占锁的场景下, 可以理解为资源占有的标识位; 在共享锁的情况下, 可以作为资源数量理解.
     *2：并且 AQS 本身是具备先进先出(FIFO)等待队列的阻塞锁和相关同步器，但是 LimitLatch 中的 Sync 则依赖自己的 AtomicLong count 作为信号量来进行判断。
     *3：用户线程通过调用LimitLatch的countUpOrAwait方法来拿到锁，如果暂时无法获取，这个线程会被阻塞到AQS的队列中。那AQS怎么知道是阻塞还是不阻塞用户线程呢？
     *其实这是由AQS的使用者来决定的，也就是内部类Sync来决定的，因为Sync类重写了AQS的tryAcquireShared()方法。它的实现逻辑是如果当前连接数count小于limit，线程能获取锁，返回 1，否则返回 -1。
     */
    public LimitLatch(long limit) {
        this.limit = limit;
        this.count = new AtomicLong(0);
        this.sync = new Sync();
    }

    /**
     * Returns the current count for the latch
     * @return the current count for latch
     */
    public long getCount() {
        return count.get();
    }

    /**
     * Obtain the current limit.
     * @return the limit
     */
    public long getLimit() {
        return limit;
    }


    /**
     * Sets a new limit. If the limit is decreased there may be a period where
     * more shares of the latch are acquired than the limit. In this case no
     * more shares of the latch will be issued until sufficient shares have been
     * returned to reduce the number of acquired shares of the latch to below
     * the new limit. If the limit is increased, threads currently in the queue
     * may not be issued one of the newly available shares until the next
     * request is made for a latch.
     *
     * @param limit The new limit
     */
    public void setLimit(long limit) {
        this.limit = limit;
    }


    /**
     * Acquires a shared latch if one is available or waits for one if no shared
     * latch is current available.
     * @throws InterruptedException If the current thread is interrupted
     */
    //判断是否能获取连接
    public void countUpOrAwait() throws InterruptedException {
        if (log.isDebugEnabled()) {
            log.debug("Counting up["+Thread.currentThread().getName()+"] latch="+getCount());
        }
        sync.acquireSharedInterruptibly(1);
    }

    /**
     * Releases a shared latch, making it available for another thread to use.
     * @return the previous counter value
     */
    public long countDown() {
        //调用AQS中的releaseShared释放锁
        sync.releaseShared(0);
        long result = getCount();
        if (log.isDebugEnabled()) {
            log.debug("Counting down["+Thread.currentThread().getName()+"] latch="+result);
        }
        //返回count数量
        return result;
    }

    /**
     * Releases all waiting threads and causes the {@link #limit} to be ignored
     * until {@link #reset()} is called.
     * @return <code>true</code> if release was done
     */
    public boolean releaseAll() {
        released = true;
        return sync.releaseShared(0);
    }

    /**
     * Resets the latch and initializes the shared acquisition counter to zero.
     * @see #releaseAll()
     */
    public void reset() {
        this.count.set(0);
        released = false;
    }

    /**
     * Returns <code>true</code> if there is at least one thread waiting to
     * acquire the shared lock, otherwise returns <code>false</code>.
     * @return <code>true</code> if threads are waiting
     */
    public boolean hasQueuedThreads() {
        return sync.hasQueuedThreads();
    }

    /**
     * Provide access to the list of threads waiting to acquire this limited
     * shared latch.
     * @return a collection of threads
     */
    public Collection<Thread> getQueuedThreads() {
        return sync.getQueuedThreads();
    }
}
